from pyspark.sql import SparkSession
from pyspark.sql import functions as F  # 避免使用import *
from pyspark.sql.functions import *
from pyspark.sql.window import Window
from pyecharts import options as opts
from pyspark.sql.functions import max as pyspark_max
from pyecharts.charts import Bar, Line, HeatMap, Pie, Tab
from pyecharts.commons.utils import JsCode
from pyecharts.globals import ThemeType
import pandas as pd
from builtins import max  # 显式导入Python内置的max函数

# 创建SparkSession
spark = SparkSession.builder \
    .appName("义乌小商品店铺人气分析-Pyecharts可视化") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()


# 读取数据
def read_data(file_path):
    df = spark.read.csv("D:\pythonspace\work_order\one\sales_data.csv", header=True, inferSchema=True)
    print("数据基本信息：")
    df.printSchema()
    print(f"数据总行数: {df.count()}")
    return df


# 数据预处理
def preprocess_data(df):
    df = df.withColumn("交易日期", to_date("交易时间"))
    df = df.withColumn("交易小时", hour("交易时间"))
    df = df.withColumn("星期几", dayofweek("交易日期"))
    df = df.withColumn("月份", month("交易日期"))
    return df


# 计算店铺人气
def calculate_shop_popularity(df):
    daily_metrics = df.groupBy("店铺名称", "交易日期") \
        .agg(
        sum("销售额").alias("日销售额"),
        sum("客流量").alias("日客流量"),
        avg("客单价").alias("日平均客单价"),
        count("交易ID").alias("日交易次数")
    )

    shop_metrics = daily_metrics.groupBy("店铺名称") \
        .agg(
        avg("日销售额").alias("平均日销售额"),
        avg("日客流量").alias("平均日客流量"),
        avg("日平均客单价").alias("平均客单价"),
        avg("日交易次数").alias("平均日交易次数"),
        sum("日客流量").alias("总客流量"),
        sum("日销售额").alias("总销售额")
    )

    max_avg_flow = shop_metrics.select(pyspark_max("平均日客流量")).collect()[0][0]
    max_avg_sales = shop_metrics.select(pyspark_max("平均日销售额")).collect()[0][0]
    max_avg_price = shop_metrics.select(pyspark_max("平均客单价")).collect()[0][0]

    shop_metrics = shop_metrics.withColumn(
        "人气评分",
        round((col("平均日客流量") / max_avg_flow * 0.4 +
               col("平均日销售额") / max_avg_sales * 0.4 +
               col("平均客单价") / max_avg_price * 0.2) * 100, 2)
    ).orderBy(col("人气评分").desc())

    return shop_metrics


# 分析商品类别与店铺人气关系
def analyze_category_popularity(df):
    category_shop = df.groupBy("商品类别", "店铺名称") \
        .agg(
        sum("销售额").alias("类别销售额"),
        sum("客流量").alias("类别客流量"),
        count("交易ID").alias("类别交易次数")
    )

    window_spec = Window.partitionBy("商品类别").orderBy(col("类别客流量").desc())
    top_shops_by_category = category_shop.withColumn("排名", rank().over(window_spec)) \
        .filter(col("排名") <= 3)

    return top_shops_by_category


# 时间趋势分析
def analyze_time_trends(df):
    daily_trends = df.groupBy("交易日期") \
        .agg(
        sum("销售额").alias("总销售额"),
        sum("客流量").alias("总客流量"),
        count("交易ID").alias("总交易次数")
    ).orderBy("交易日期")

    weekday_trends = df.groupBy("星期几") \
        .agg(
        sum("销售额").alias("总销售额"),
        sum("客流量").alias("总客流量"),
        count("交易ID").alias("总交易次数")
    ).orderBy("星期几")

    hourly_trends = df.groupBy("交易小时") \
        .agg(
        sum("销售额").alias("总销售额"),
        sum("客流量").alias("总客流量"),
        count("交易ID").alias("总交易次数")
    ).orderBy("交易小时")

    return daily_trends, weekday_trends, hourly_trends


# Pyecharts可视化函数
def create_shop_ranking_chart(shop_metrics):
    """店铺人气柱状图"""
    df_pd = shop_metrics.toPandas()
    x_data = df_pd["店铺名称"].tolist()
    y_data = df_pd["人气评分"].tolist()

    return (
        Bar(init_opts=opts.InitOpts(theme=ThemeType.LIGHT, width="1000px", height="600px"))
        .add_xaxis(x_data)
        .add_yaxis("人气评分", y_data, itemstyle_opts=opts.ItemStyleOpts(color="#4e79a7"))
        .set_global_opts(
            title_opts=opts.TitleOpts(title="店铺人气排名"),
            xaxis_opts=opts.AxisOpts(axislabel_opts=opts.LabelOpts(rotate=45)),
            tooltip_opts=opts.TooltipOpts(trigger="axis", axis_pointer_type="shadow")
        )
    )


def create_daily_trend_chart(daily_trends):
    """每日销售额与客流量趋势图"""
    df_pd = daily_trends.toPandas()
    x_data = df_pd["交易日期"].astype(str).tolist()

    return (
        Line(init_opts=opts.InitOpts(width="1000px", height="600px"))
        .add_xaxis(x_data)
        .add_yaxis(
            "销售额",
            df_pd["总销售额"].tolist(),
            yaxis_index=0,
            itemstyle_opts=opts.ItemStyleOpts(color="#e15759")
        )
        .add_yaxis(
            "客流量",
            df_pd["总客流量"].tolist(),
            yaxis_index=1,
            itemstyle_opts=opts.ItemStyleOpts(color="#59a14f")
        )
        .extend_axis(
            yaxis=opts.AxisOpts(
                name="客流量",
                type_="value",
                position="right",
            )
        )
        .set_global_opts(
            title_opts=opts.TitleOpts(title="每日销售额与客流量趋势"),
            xaxis_opts=opts.AxisOpts(axislabel_opts=opts.LabelOpts(rotate=45)),
            tooltip_opts=opts.TooltipOpts(trigger="axis")
        )
    )


def create_weekday_heatmap(weekday_trends, hourly_trends):
    """星期×小时客流量热力图"""
    hourly_pd = hourly_trends.toPandas()
    weekday_pd = weekday_trends.toPandas()

    weekday_map = {1: "周一", 2: "周二", 3: "周三", 4: "周四", 5: "周五", 6: "周六", 7: "周日"}
    heatmap_data = []

    # 计算每个时段的客流量（星期×小时）
    total_weekday_flow = weekday_pd["总客流量"].sum()
    for weekday_num in range(1, 8):
        weekday_flow = weekday_pd[weekday_pd["星期几"] == weekday_num]["总客流量"].values[0]
        weekday_ratio = weekday_flow / total_weekday_flow if total_weekday_flow != 0 else 0

        for hour in range(0, 24):
            hour_flow = hourly_pd[hourly_pd["交易小时"] == hour]["总客流量"].values[0]
            flow_value = int(hour_flow * weekday_ratio)
            heatmap_data.append([weekday_map[weekday_num], hour, flow_value])

    # 使用显式导入的Python内置max函数
    max_flow = max([v[2] for v in heatmap_data]) if heatmap_data else 0

    return (
        HeatMap(init_opts=opts.InitOpts(width="1000px", height="600px"))
        .add_xaxis([weekday_map[i] for i in range(1, 8)])
        .add_yaxis(
            "小时",
            [f"{i}时" for i in range(0, 24)],
            heatmap_data,
            label_opts=opts.LabelOpts(is_show=True, formatter=JsCode("function(params){return params.data[2];}")),
        )
        .set_global_opts(
            title_opts=opts.TitleOpts(title="客流量分布热力图（星期×小时）"),
            visualmap_opts=opts.VisualMapOpts(
                min_=0,
                max_=max_flow,
                range_text=["高", "低"],
                orient="horizontal",
                pos_right="center"
            ),
        )
    )


def create_category_pie_chart(category_data):
    """商品类别销售额占比饼图"""
    df_pd = category_data.groupBy("商品类别").agg(sum("类别销售额").alias("总销售额")).toPandas()
    data = [list(z) for z in zip(df_pd["商品类别"].tolist(), df_pd["总销售额"].tolist())]

    return (
        Pie(init_opts=opts.InitOpts(width="1000px", height="600px"))
        .add(
            "销售额占比",
            data,
            radius=["40%", "70%"],
            label_opts=opts.LabelOpts(formatter="{b}: {d}%")
        )
        .set_global_opts(
            title_opts=opts.TitleOpts(title="商品类别销售额占比"),
            legend_opts=opts.LegendOpts(orient="vertical", pos_left="left")
        )
    )


# 主函数：生成HTML看板
def main():
    print("===== 开始义乌小商品店铺人气分析 =====")

    # 数据处理
    sales_data = read_data("sales_data.csv")
    processed_data = preprocess_data(sales_data)

    # 计算核心指标
    shop_popularity = calculate_shop_popularity(processed_data)
    category_popularity = analyze_category_popularity(processed_data)
    daily_trends, weekday_trends, hourly_trends = analyze_time_trends(processed_data)

    # 创建可视化图表
    tab = Tab(page_title="义乌小商品店铺分析看板")

    # 添加图表到标签页
    tab.add(create_shop_ranking_chart(shop_popularity), "店铺人气排名")
    tab.add(create_daily_trend_chart(daily_trends), "每日趋势分析")
    tab.add(create_weekday_heatmap(weekday_trends, hourly_trends), "时段分布热力图")
    tab.add(create_category_pie_chart(category_popularity), "商品类别分析")

    # 保存为HTML文件
    tab.render("义乌小商品店铺分析看板.html")
    print("\n===== 分析完成 =====")

    # 停止SparkSession
    spark.stop()


if __name__ == "__main__":
    main()